package cn.itcast.model.base

import java.util.Properties

import cn.itcast.model.bean.HBaseMeta
import org.apache.commons.lang3.StringUtils
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

trait BaseModel {
  //fghhf
  //指定应用的名字
  def setAppName(): String
  // 提醒儿子进行导包操作
  def importSparkEnv()
  // 指定标签的ID
  def setTagId(): Int
  //真正进行标签计算的方法
  def computeTag(fiveRuleDF: DataFrame, hbaseSource: DataFrame): DataFrame

 val spark: SparkSession = SparkSession.builder()
    .appName(setAppName())
    .master("local[*]")
    .getOrCreate()
  //2. 导包
  import spark.implicits._
  import org.apache.spark.sql.functions._
  //3. 加载MySQL数据源
  def loadMySQLSource(): DataFrame = {
    val url = "jdbc:mysql://bd001:3306/tags_new?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&user=root&password=123456"
    val table = "tbl_basic_tag"
    val properties = new Properties()
    val mysqlSource: DataFrame = spark.read.jdbc(url, table, properties)
    mysqlSource
  }
  def getFourRuleData(mysqlSource:DataFrame):Map[String,String]={
    val result: Map[String, String] = mysqlSource.select('rule).where("id=" + setTagId()).rdd
      .map(row => {
        val str = row.getAs[String]("rule")
        val arr: Array[String] = str.split("##")
        val tuples: Array[(String, String)] = arr.map(word => {
          val tmpArr = word.split("=")
          tmpArr(0) -> tmpArr(1)
        })
        val map = tuples.toMap
        map
      }).collect()(0)
    result
  }
  //   2. 获取5级规则数据: 将获取到的数据(id,rule)=>TagRule
  def getFiveRuleData(mysqlSource: DataFrame): DataFrame = {
    mysqlSource.select('id, 'rule).where("pid = " + setTagId)
  }
  //5. 加载HBase中的原始数据
  def loadHBaseSource(param: Map[String, String]): DataFrame = {
    spark.read
      .format("cn.itcast.model.utils.HBaseSource")
      .options(param)
      .load()
  }
  //6. 开始进行职业标签计算(需要儿子实现)

  //7. 将数据落地到HBase,需要进行标签的合并操作
  //   1. 读取之前的历史标签
  def loadOldDF(): DataFrame = {
    val oldDF: DataFrame = spark.read
      //      .format("自定义数据源的全路径")
      .format("cn.itcast.model.utils.HBaseSource")
      //      .option("参数名", "参数值")
      .option(HBaseMeta.ZKHOSTS, "192.168.10.20")
      .option(HBaseMeta.ZKPORT, "2181")
      .option(HBaseMeta.HBASETABLE, "test_bs")
      .option(HBaseMeta.FAMILY, "detail")
      .option(HBaseMeta.SELECTFIELDS, "userId,tagIds")
      .load()
    oldDF
  }
  //   2. 将新老标签数据进行合并操作
  def mergeTag(newDF: DataFrame, oldDF: DataFrame): DataFrame = {
    //   2. 将新老标签数据进行合并操作
    // (右表, 关联条件, 关联类型) 全连接: 包含新表和老表的所有用户信息.
    val joinDF: DataFrame = newDF.join(oldDF, newDF.col("userId") === oldDF.col("userId"), "full")
    // 对新老数据进行合并操作.
    // 定义一个自定义函数,实现标签的合并操作.
    val mergeTag = udf((newTag: String, oldTag: String) => {
      // 如果进行if else 判断,一定要确保最终有返回值.
      // 我们可以先看看新老数据中是否有值,如果出现空值,直接返回即可
      if (StringUtils.isBlank(newTag)) {
        // 判断新数据是否为空
        oldTag
      } else if (StringUtils.isBlank(oldTag)) {
        // 判断老数据是否为空
        newTag
      } else if (StringUtils.isNotBlank(newTag) && StringUtils.isNotBlank(oldTag)) {
        // 新老数据都不为空
        //开始合并 老数据,新数据
        val temStr: String = oldTag + "," + newTag
        // 有可能会出现重复标签,需要进行去重 2,22,45,22
        val set: Set[String] = temStr.split(",").toSet
        val mergeTagStr: String = set.mkString(",")
        mergeTagStr
      } else {
        ""
      }
    })
    val result: DataFrame = joinDF.select(
      // 去新表中获取用户ID信息,如果找不到,就去老表中获取
      when(newDF.col("userId").isNotNull, newDF.col("userId"))
        .when(newDF.col("userId").isNull, oldDF.col("userId"))
        .as("userId"),
      mergeTag(newDF.col("tagIds"), oldDF.col("tagIds")).as("tagIds")
    )
    result
  }
  //   3. 将合并后的结果保存到HBase
  def saveResult(result: DataFrame) = {
    result.write
      //      .format("自定义数据落地的数据源")
      .format("cn.itcast.model.utils.HBaseSource")
      //      .option("参数名", "参数值")
      .option(HBaseMeta.ZKHOSTS, "192.168.10.20")
      .option(HBaseMeta.ZKPORT, "2181")
      .option(HBaseMeta.HBASETABLE, "test_bs")
      .option(HBaseMeta.FAMILY, "detail")
      .option(HBaseMeta.SELECTFIELDS, "userId,tagIds")
      .save()
  }
  def executeModel():DataFrame = {
    //    1. 创建SparkSession对象
    //2. 导包
    //3. 加载MySQL数据源
    val mysqlSource: DataFrame = loadMySQLSource()
    //4. 开始获取4/5级规则数据
    //   1. 获取4级规则数据: 将获取到的数据=>HBaseMeta
    val fourRuleMap: Map[String, String] = getFourRuleData(mysqlSource)
    //   2. 获取5级规则数据: 将获取到的数据(id,rule)=>TagRule
    val fiveRuleDF: DataFrame = getFiveRuleData(mysqlSource)
    //5. 加载HBase中的原始数据
    val hbaseSource: DataFrame = loadHBaseSource(fourRuleMap)
    //6. 开始进行职业标签计算
    val newDF: DataFrame = computeTag(fiveRuleDF, hbaseSource)
    newDF
  }
    //   3. 将合并后的结果保存到HBase
    def saveData(newDF: DataFrame)={
      //7. 将数据落地到HBase,需要进行标签的合并操作
      //   1. 读取之前的历史标签
      val oldDF: DataFrame = loadOldDF()
      //   2. 将新老标签数据进行合并操作
      val result: DataFrame = mergeTag(newDF, oldDF)
      //   3. 将合并后的结果保存到HBase
      saveResult(result)
    }
}
